package com.zqh.kafka.demo.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;

public class MyConsumer1 {

    public static void main(String[] args) {

        Map<String, Object> configs = new HashMap<>();
        // 添加配置参数  可以使用常量代替手写输入
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "centos122:9092"); // 服务地址
        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class); // 指定key的序列化类
        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); // 指定value的序列化类

        configs.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer_demo"); // 配置groupId(消费组ID)

        // 如果找不到当前消费者的有效偏移量，则自动重置到开始
        // latest 表示直接重置到消息偏移量的最后一个
        configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // 创建消费者
        KafkaConsumer<Integer, String> consumer = new KafkaConsumer<Integer, String>(configs);

        // 先订阅，再消费
        consumer.subscribe(Arrays.asList("topic_1"));

        while (true) {
            // 批量从主题的分区拉取消息，每间隔3秒拉取一次
            ConsumerRecords<Integer, String> consumerRecords = consumer.poll(3_000);

            // 遍历本次从主题的分区拉取的批量消息
            consumerRecords.forEach(new Consumer<ConsumerRecord<Integer, String>>() {
                @Override
                public void accept(ConsumerRecord<Integer, String> record) {
                    System.out.println(record.topic());
                    System.out.println(record.partition());
                    System.out.println(record.offset());
                    System.out.println(record.key());
                    System.out.println(record.value());
                }
            });
        }

//        // 批量从主题的分区拉取消息，每间隔3秒拉取一次
//        ConsumerRecords<Integer, String> consumerRecords = consumer.poll(3_000);
//
//        // 遍历本次从主题的分区拉取的批量消息
//        consumerRecords.forEach(new Consumer<ConsumerRecord<Integer, String>>() {
//            @Override
//            public void accept(ConsumerRecord<Integer, String> record) {
//                System.out.println(record.topic());
//                System.out.println(record.partition());
//                System.out.println(record.offset());
//                System.out.println(record.key());
//                System.out.println(record.value());
//                System.out.println("***********************");
//            }
//        });

        // 关闭consumer
//        consumer.close();

    }

}
